package com.zc.redisson.delayedQueue.manager;


import com.zc.util.SpringUtils;
import com.zc.redisson.delayedQueue.handler.DelayMessageHandler;
import com.zc.redisson.delayedQueue.message.DelayMessage;
import com.zc.redisson.delayedQueue.message.DelayMessageType;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @description 基于redisson实现的延时消息管理器
 */
@Component
@ConditionalOnWebApplication
public class RedissonDelayMessageManager implements DelayMessageManager {

    private final Logger log = LoggerFactory.getLogger(RedissonDelayMessageManager.class);

    private final RBlockingQueue<DelayMessage<?>> rBlockingQueue;

    private final RDelayedQueue<DelayMessage<?>> rDelayedQueue;

    private final Executor emsThreadPoolTaskExecutor;


    public RedissonDelayMessageManager(RedissonClient redissonClient, @Qualifier("emsThreadPoolTaskExecutor") Executor emsThreadPoolTaskExecutor) {
        rBlockingQueue = redissonClient.getBlockingDeque("redisson-delay-message-queue");
        rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
        this.emsThreadPoolTaskExecutor = emsThreadPoolTaskExecutor;
    }


    private final Map<DelayMessageType, DelayMessageHandler<?>> handlerMap = new ConcurrentHashMap<>(16);

    @Override
    public void add(DelayMessage<?> message) {
        if (rDelayedQueue.contains(message)) {
            return;
        }
        log.info("redisson-delay-message-queue,add message = {}", message);
        rDelayedQueue.offer(message, message.getExpire(), message.getTimeUnit());
    }

    @Override
    public boolean remove(DelayMessage<?> message) {
        return rDelayedQueue.remove(message);
    }

    @Override
    public void destroy() {
        rDelayedQueue.destroy();
    }

    @Override
    public void afterPropertiesSet() {
        Arrays.stream(DelayMessageType.values()).forEach(delayMessageType -> handlerMap.put(delayMessageType, SpringUtils.getBean(delayMessageType.getHandler())));
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    // 获取并移除此队列的头部，在元素变得可用之前一直等待 。queue的长度 == 0 的时候，阻塞等待
                    DelayMessage delayMessage = rBlockingQueue.poll(2, TimeUnit.SECONDS);;
//                    log.info("redisson-delay-message-queue,consume message = {}", delayMessage);
                    if (delayMessage == null) {
                        continue;
                    }
                    DelayMessageHandler<?> delayMessageHandler = handlerMap.get(delayMessage.getType());
                    emsThreadPoolTaskExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            delayMessageHandler.handle(delayMessage);
                        }
                    });
                } catch (InterruptedException e) {
                   log.error("redisson-delay-message-queue，Interrupted",e);
                    afterPropertiesSet();
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }
}
